package com.xiaoming;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import org.apache.curator.RetryPolicy;
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.CuratorFrameworkFactory;
import org.apache.curator.retry.ExponentialBackoffRetry;
import org.apache.zookeeper.CreateMode;
import org.apache.zookeeper.Watcher;
import org.apache.zookeeper.ZKUtil;
import org.apache.zookeeper.data.Stat;

import java.io.InputStream;
import java.util.*;

public class ZKBuilder {
    private  CuratorFramework client = null;
    private static ZKBuilder zkBuilder;
    private String rootName = "/rpc-homework1";
    private Map<String, Address> serverMap = new HashMap<>();

    public static ZKBuilder build(WatcherCallback callback,Integer sessionTimeoutMs) throws Exception {
        if (zkBuilder == null) {
            zkBuilder = new ZKBuilder();
            zkBuilder.init(sessionTimeoutMs);
            if (callback != null) {
                zkBuilder.registerWatcher(callback);
            }
        }
        return zkBuilder;
    }

    public void createNode(String name, Object obj) throws Exception {
        String nodePath = rootName + name;
        Stat stat = client.checkExists().forPath(nodePath);
        if (stat == null) {
            client.create().creatingParentsIfNeeded().withMode(CreateMode.EPHEMERAL_SEQUENTIAL).forPath(nodePath, JSON.toJSONBytes(obj));
        }
    }

    public Map<String, Address> getServerMap() throws Exception {
        List<String> children = client.getChildren().forPath(rootName);
        for (String child : children) {
            byte[] bytes = client.getData().forPath(rootName + "/"+ child);
            JSONObject json = JSONObject.parseObject(new String(bytes));
            Address address = new Address();
            address.setHost(json.getString("host"));
            address.setPort(json.getInteger("port"));
            serverMap.put(child, address);
        }
        return serverMap;
    }

    private void registerWatcher(WatcherCallback callback) throws Exception {
        Watcher watcher = watchedEvent -> {
            if (watchedEvent.getType() == Watcher.Event.EventType.NodeChildrenChanged) {
                try {
                    List<String> children = client.getChildren().forPath(rootName);
                    for (String child : children) {
                        String serverName = rootName + "/" +child;
                        byte[] bytes = client.getData().forPath(serverName);
                        JSONObject json = JSONObject.parseObject(new String(bytes));
                        Address address = new Address();
                        address.setHost(json.getString("host"));
                        address.setPort(json.getInteger("port"));
                        if (!serverMap.containsKey(child)) {
                            callback.change(address,"ADD");
                            serverMap.put(child, address);
                        }
                    }
                    Iterator<String> iterator = serverMap.keySet().iterator();
                    while (iterator.hasNext()){
                        String next = iterator.next();
                        if(!children.contains(next)){
                            callback.change(serverMap.get(next),"DELETE");
                            iterator.remove();
                        }
                    }
                    System.out.println(String.format("节点列表改变为：%s", children));
                } catch (Exception e) {
                    e.printStackTrace();
                }finally {
                    try {
                        registerWatcher(callback);
                    } catch (Exception e) {
                        e.printStackTrace();
                    }
                }
            }
        };
        client.getChildren().usingWatcher(watcher).forPath(rootName);
    }

    private void init(Integer sessionTimeoutMs) throws Exception {
        InputStream in = ZKUtil.class.getClassLoader().getResourceAsStream("zookeeper.properties");
        Properties properties = new Properties();
        properties.load(in);
        String namespace = properties.getProperty("zookeeper.namespace");
        RetryPolicy retryPolicy = new ExponentialBackoffRetry(1000, 3);
        String zkAddress = properties.getProperty("zookeeper.server.address");
        client = CuratorFrameworkFactory.builder()
                .connectString(zkAddress)
                .sessionTimeoutMs(sessionTimeoutMs)
                .connectionTimeoutMs(3000)
                .retryPolicy(retryPolicy)
                .namespace(namespace)
                .build();
        client.start();
        if (client.checkExists().forPath(rootName) == null) {
            client.create().withMode(CreateMode.PERSISTENT).forPath(rootName);
        }
    }

    private ZKBuilder() {
    }
}
